package com.fucheng.transport;

import com.fucheng.common.RpcRequest;
import com.fucheng.common.RpcResponse;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.concurrent.atomic.AtomicReference;

/**
 * Netty 客户端，用于发送 RpcRequest 并同步接收 RpcResponse
 */
public class NettyRpcClient {

    public RpcResponse send(RpcRequest rpcRequest, String host, int port) throws Exception {
        AtomicReference<RpcResponse> responseRef = new AtomicReference<>();

        EventLoopGroup group = new NioEventLoopGroup();

        try{
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<Channel>() {

                        @Override
                        protected void initChannel(Channel channel) throws Exception {

                            ChannelPipeline p = channel.pipeline();
                            p.addLast(new RpcEncoder()); // Object -> ByteBuf
                            p.addLast(new RpcDecoder()); // ByteBuf -> Object
                            p.addLast(new SimpleClientHandler(responseRef));
                        }
                    });

            ChannelFuture f = bootstrap.connect(host, port).sync();
            f.channel().writeAndFlush(rpcRequest).sync(); // 发送请求

            // 等待服务端处理完关闭连接
            f.channel().closeFuture().sync();

            return responseRef.get();
        } finally {
            group.shutdownGracefully();
        }
    }
}
